package plgnet

import (
	"dgo/MyProj/PrjConf"
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/dnet"
	"dgo/goutils/utils"
	"dgo/plg/plgbase"
	"fmt"
	"gitee.com/ymofen/supervalue"
	"net/http"
)

type PlgNetTCPSvr struct {
	plgbase.PlgBase
	monitor_recv_flag int8 // 监测收到的数据
	monitor_input     *utils.MonitorSpeedSizeRec
	msglst            *utils.SyncCycleList
	tcpsvr            *dnet.TcpServer
	outputEx          *PlgBaseInnerEx
	msgShowLvl        int8
	lastwarn          string
}

func (this *PlgNetTCPSvr) Close() error {
	this.tcpsvr.DisconnectAll()
	this.tcpsvr.Stop()
	this.tcpsvr = nil
	this.PlgBase.Close()
	utils.Monitor.RemoveMonitorRec(this.monitor_input)
	return nil
}

func (this *PlgNetTCPSvr) OnHttpStatus(rec *PlgHttpRec) (rval interface{}, errcode int, err error) {
	itype := rec.QryAsInt("type", 0)
	if itype == 99 {
		json := wrapper.NewSVObject()

		json.SetKeyValues("plugid", this.PlugId, "uuid", PrjConf.APP_UUID,
			"input-size", this.monitor_input.Info(),
			"addr", this.tcpsvr.GetAddr(),
			"online", this.tcpsvr.GetOnlineCnt(),
			"sending-max-cache", this.tcpsvr.GetMaxSendingQueueSize(),
			"checktimeout-active", utils.HumanDuration(this.tcpsvr.GetActiveTimeOutInterval()),
			"maxonline", this.tcpsvr.GetMaxOnline(),
			"active", this.tcpsvr.GetActive(),
			"output", this.outputEx.StatusString(),
			"lastwarn", this.lastwarn,
		)

		if this.InputEx != nil {
			json.SetKeyValues("input-ex", this.InputEx.StatusString())
		}
		json.SetKeyValues("now", utils.NowString())

		if rec.QryAsInt("json", 0) == 1 {

			PrjHttpUtils.ResponseAsJSON(rec.RespW, json.AsJSONStringEx(false, true))
		} else {
			json.SetKeyValue("opera", fmt.Sprintf("【%s】-【%s】-【%s】",
				fmt.Sprintf("<a href=\"%s\" target=\"_blank\">关闭显示</a>", "status?type=97&flag=0"),
				fmt.Sprintf("<a href=\"%s\" target=\"_blank\">文本显示</a>", "status?type=97&flag=1"),
				fmt.Sprintf("<a href=\"%s\" target=\"_blank\">HEX显示</a>", "status?type=97&flag=2"),
			))
			PrjHttpUtils.JSON2TableRec(rec.RespW, "", json.AsJSONString(false))
		}
	} else if itype == 98 {
		rec.ResponseSyncListNEx(this.msglst)
	} else if itype == 97 {
		this.monitor_recv_flag = int8(rec.QryAsInt("flag", 0))
		rec.ResponseJSONValues("code", 0, "flag", this.monitor_recv_flag)
	} else if itype == 96 {
		lvl := int8(rec.IntByName("lvl", -1))
		if lvl != -1 {
			this.msgShowLvl = lvl
		}
		rec.ResponseJSONValues("code", 0, "lvl", this.msgShowLvl)
	} else if itype == 1 {
		tcpsvr := this.tcpsvr
		if tcpsvr.MonitorFlag == 0 {
			tcpsvr.MonitorFlag = 1
		}

		lst := wrapper.NewSVArray()
		cnt := tcpsvr.RangeOnlineList(func(conn *dnet.TcpServerClientConnection) bool {
			itm := lst.AppendObject()
			itm.SetKeyValues("from", conn.GetRemoteAddr(),
				"recv-info", conn.RecvMonitor.Info(),
				"send-info", conn.SendMonitor.Info(),
				"start-t", utils.DateTimeString3(conn.GetConnectedTime()),
				"timeout-t", fmt.Sprintf("%.2f(s)", conn.GetTimeOut().Seconds()),
				"cache-size", conn.GetCacheInfo(),
				"lastwarning", conn.LastWarning)
			return true
		})

		if cnt == 0 {
			http.Error(rec.RespW, "503 none connection!", 503)
			return
		}

		if rec.QryAsInt("json", 0) == 1 {
			PrjHttpUtils.ResponseAsJSON(rec.RespW, lst.AsJSONStringEx(false, false))
		} else {
			conf_title := wrapper.NewSVObject()
			conf_title.SetKeyValues("from", "连接地址", "recv-info", "接收统计",
				"send-info", "发送统计", "start-t", "连接时间", "timeout-t", "交互延迟(s)", "cache-size", "发送队列", "lastwarning", "信息")
			PrjHttpUtils.JSON2Table(rec.RespW, conf_title.AsJSONString(false), lst.AsJSONString(false))
		}
	} else {
		var sb utils.BytesBuilder
		PrjHttpUtils.AppendIndexBody(&sb, "status", "status?type=99")
		PrjHttpUtils.AppendIndexBody(&sb, "msglist", "status?type=98&n=100&searchval=")
		PrjHttpUtils.AppendIndexBody(&sb, "monitor-recv-flag", "status?type=97&flag=0")
		PrjHttpUtils.AppendIndexBody(&sb, "msgshowlvl", "status?type=96&lvl=")
		PrjHttpUtils.AppendIndexBody(&sb, "client-lst", "status?type=1&n=100&searchval=")
		PrjHttpUtils.ResponseBodyHtml(rec.RespW, sb.String())
	}
	return

}

/*
*

	{
		"id":"bnc-disp",
		"type":"net.tcpsvr",
		"conf":{
			"input":{"id":"","topic":"topic", "desc":"如果指定了input则将订阅到的数据通过tcpsvr进行发送"},
			"addr":":5002",
			"maxonline":0,
			"timeout-secs":60
		}
	}

	{
	  "input":{        // 如果指定了input则将订阅到的数据通过tcpsvr进行发送
	     "id":"",
	     "topic":"topic"
	   },
	   "addr":":5002",
	   "monitor-flag":1,
	   "monitor-recv-flag":1,
	   "maxonline":1,
	   "timeout-secs":10,    // 10秒内无数据交互, 会断线
	   "status_o":{
	       "id":"xxx",
	       "topic":"/cmd/status",
	       "interval":1
	   },
	   "output":{
	      "id":"xx",
	      "topic":"xxxx"
	   }
	}
*/
func (this *PlgNetTCPSvr) ReloadConf(conf *wrapper.SuperValue) error {
	if this.outputEx == nil {
		this.outputEx = PlgNewInnerEx()
		this.outputEx.ConfigOwnerPlug(this)
	}

	if this.msglst == nil {
		this.msglst = utils.NewSyncCycleList()
		this.msgShowLvl = 2
	}

	this.msgShowLvl = conf.Int8ByName("msgshowlvl", this.msgShowLvl)

	if this.monitor_input == nil {
		this.monitor_input = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitor(this.monitor_input)
	}

	if this.tcpsvr == nil {
		this.tcpsvr = dnet.NewTcpServerEx(this.PlugId)
		this.tcpsvr.OnDisEvent = func(sender *dnet.TcpServerClientConnection) {
			this.msglst.LogMsg("[%s][%s]:%s", utils.NowString(), sender.GetRemoteAddr(), sender.GetDisReason())
		}
		this.tcpsvr.OnListenSucc = func(sender *dnet.TcpServer) {
			this.LogInfoMessage(2, "[%s]侦听成功", sender.GetAddr())
		}
		this.tcpsvr.OnErr = func(sender *dnet.TcpServer, err error) {
			this.LogInfoMessage(2, "[%s]侦听成功", sender.GetAddr())
		}

		this.tcpsvr.OnRecvBufEvent = func(sender *dnet.TcpServerClientConnection, buf []byte) {
			if this.monitor_recv_flag == 1 {
				this.msglst.LogMsg("[%s][%s]:%s", utils.NowString(), sender.GetRemoteAddr(), []byte(buf))
			} else if this.monitor_recv_flag == 2 {
				this.msglst.LogMsg("[%s][%s]:%s", utils.NowString(), sender.GetRemoteAddr(), utils.BufToHexStr(buf, 0, ""))
			}
			this.outputEx.OutputBufferEx(buf, sender, nil)
		}
		this.tcpsvr.UseQueueSend = true
	}
	this.tcpsvr.MonitorFlag = conf.IntByName("monitor-flag", 1)
	this.tcpsvr.ConfigMaxSendingQueueSize(conf.IntByName("sending-max-cache", 1024*100))

	this.monitor_recv_flag = conf.Int8ByName("monitor-recv-flag", 0)

	conf.CheckExistsIntFunc("maxonline", func(val int64) {
		this.tcpsvr.SetMaxOnline(uint32(val))
	})

	conf.CheckExistsIntFunc("timeout-secs", func(val int64) {

		//this.tcpsvr.ConfigCheckTimeoutInterval(5)
		this.tcpsvr.ConfigKickTimeOut(int(val))
	})

	conf.CheckStrFunc("addr", func(strv string) {
		this.tcpsvr.SetAddress(strv)
		if !this.tcpsvr.GetActive() {
			this.tcpsvr.Start()
		}
	})

	this.outputEx.ReloadConf(conf.ValueByName("output"))

	statusOConf := conf.ValueByName("status-o")
	if statusOConf == nil {
		statusOConf = conf.ValueByName("status_o")
	}

	this.ReloadStatusOConf(statusOConf, func() {
		this.StatusOutputEx.SendStatusFunc(func(json *wrapper.SuperValue) {
			json.SetKeyValues("plugid", this.PlugId, "online", this.tcpsvr.GetOnlineCnt())
		})
	}, func() {
		//utils.Monitor.AddMonitorRec(fmt.Sprintf("plugin_%d_recv", this.PlugDNA), &this.recvMonitor)
	}, func() {
		//utils.Monitor.RemoveMonitorRec(fmt.Sprintf("plugin_%d_recv", this.PlugDNA))
	})

	this.ReloadInputConf(conf.ValueByName("input"), this.OnInputSubscribeCallback)
	return nil
}

func (this *PlgNetTCPSvr) OnObjectNotify(sender interface{}, notifytype int, data interface{}) (rval interface{}, err error) {
	if notifytype == utils.NOTIFY_TYPE_SETCONF {
		if conf, ok := data.(*wrapper.SuperValue); ok {
			err = this.ReloadConf(conf)
		} else {
			err = fmt.Errorf("非法的JSON配置")
		}
	} else if notifytype == utils.NOTIFY_TYPE_RELEASE {
		this.ReleaseFlag = 1
		this.Release()

		this.outputEx.Release()
		if this.tcpsvr != nil {
			this.tcpsvr.DisconnectAll()
			this.tcpsvr.Stop()
			this.tcpsvr = nil
		}
	}
	return
}

func (this *PlgNetTCPSvr) DispatchData(channelid string, sender interface{}, data interface{}, tag interface{}) (err error) {
	if this.ReleaseFlag == 1 {
		return nil
	}
	tcpsvr := this.tcpsvr
	if tcpsvr != nil {
		if buf, ok := data.([]byte); ok {
			this.monitor_input.DeltaBuf(buf)
			tcpsvr.Send2All(buf)
		}
	}
	return nil
}

func (this *PlgNetTCPSvr) OnInputSubscribeCallback(sender interface{}, channelid string, subscribeid interface{}, data interface{}, tag interface{}, isBreak *bool) bool {
	if this.ReleaseFlag == 1 {
		return true
	}
	tcpsvr := this.tcpsvr
	if tcpsvr != nil {
		if buf, ok := data.([]byte); ok {
			this.monitor_input.DeltaBuf(buf)
			tcpsvr.Send2All(buf)
		}
	}
	return true
}

/*
msglvl: 值越大信息约重要
*/
func (this *PlgNetTCPSvr) LogInfoMessage(msglvl int8, s string, args ...interface{}) {
	if msglvl < this.msgShowLvl {
		return
	}
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_INFO, 1, s, args...)
}

func (this *PlgNetTCPSvr) LogWarnMessage(msglvl int8, s string, args ...interface{}) {
	if msglvl < this.msgShowLvl {
		return
	}
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_WARN, 1, s, args...)
}

func CreatePlgTCPSvrFunc(args ...interface{}) (rval interface{}, err error) {
	obj := &PlgNetTCPSvr{}
	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	obj.RealPlg = obj
	rval = obj
	return
}
